package com.atguigu.chapter05.transform;

import com.atguigu.bean.WaterSensor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.ArrayList;

/**
 * Author: Pepsi
 * Date: 2023/7/28
 * Desc:
 */
public class Flink05_sum_min_minBy {
    public static void main(String[] args) {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port",1000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);

        ArrayList<WaterSensor> waterSensors = new ArrayList<>();
        waterSensors.add(new WaterSensor("sensor_1",1607527992000L, 20));
        waterSensors.add(new WaterSensor("sensor_1",1607527994000L, 50));
        waterSensors.add(new WaterSensor("sensor_1",1607527996000L, 10));
        waterSensors.add(new WaterSensor("sensor_2",1607527993000L, 10));
        waterSensors.add(new WaterSensor("sensor_2",1607527995000L, 30));

        DataStreamSource<WaterSensor> waterSensorDataStreamSource = env.fromCollection(waterSensors);

        waterSensorDataStreamSource
                .keyBy(new KeySelector<WaterSensor, String>() {
                    @Override
                    public String getKey(WaterSensor value) throws Exception {
                        return value.getId();
                    }
                })
//                .min("vc")  // 只计算指定字段的最小值，其他字段的数据会保留最初第一个数据的值
                .minBy("vc") // 会返回包含字段最小值的整条数据
                .print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}
